package com.tmsb.ucenter;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.tmsb.pojo.MysqlBinlog;
import com.tmsb.pojo.UcenterUser;
import com.tmsb.sink.DauKeyStateSink;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.Const;
import com.tmsb.utils.Util;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.contrib.streaming.state.PredefinedOptions;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Description: 用ucenter的user来获取日活、省日活、日注册、省日注册这四个实时指标
 * Version:1.0.0
 *
 * @author xiekb
 * @date 2020/6/29
 */
public class DailyActivateUnits {
    private static final OutputTag<UcenterUser> DAILY_ACTIVATE_UNITS = new OutputTag<UcenterUser>("DailyActivateUnits") {};
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//      env.setStateBackend(new FsStateBackend("file:///C:\\Users\\home\\Desktop\\mall\\orderGoods"));
        RocksDBStateBackend rocksDBStateBackend = new RocksDBStateBackend(
                ConfUtil.getDauCheckpointUri(), true);
        rocksDBStateBackend.setNumberOfTransferThreads(3);
        rocksDBStateBackend.setPredefinedOptions(
                PredefinedOptions.SPINNING_DISK_OPTIMIZED_HIGH_MEM);
        rocksDBStateBackend.enableTtlCompactionFilter();
        env.setStateBackend(rocksDBStateBackend);

        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.enableCheckpointing(ConfUtil.getCheckpointInterval(), CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(ConfUtil.getCheckpointMinPauseBetween());
        env.setParallelism(ConfUtil.getParallelism());

        //重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
                ConfUtil.getfailureRate(), org.apache.flink.api.common.time.Time.of(ConfUtil.getfailureInterval(), TimeUnit.MINUTES),
                org.apache.flink.api.common.time.Time.of(ConfUtil.getdelayInterval(), TimeUnit.SECONDS)
        ));

        // source

        DataStream<String> source = env
                .addSource(new FlinkKafkaConsumer<>(ConfUtil.getTopicOfUcenter(), new SimpleStringSchema(), ConfUtil.getKafkaProperties()))
                .uid("kafka-source");

        SingleOutputStreamOperator<Object> data = source
                .map(s -> JSON.parseObject(s, MysqlBinlog.class))
                .uid("map-mysqlbinlog")
                .process(new ProcessFunction<MysqlBinlog, Object>() {

                    @Override
                    public void processElement(MysqlBinlog mysqlBinlog, Context context,
                                               Collector<Object> collector) throws Exception {
                if (Const.MysqlDatabase.UCENTER.toString().equalsIgnoreCase(mysqlBinlog.getDatabase())) {
                    if (Const.MysqlTable.USER.toString().equalsIgnoreCase(mysqlBinlog.getTable())) {
                        UcenterUser ucenterUser = JSON.parseObject(mysqlBinlog.getData(), UcenterUser.class);
                        if(ucenterUser==null){
                            return;
                        }
                        ucenterUser.setType(mysqlBinlog.getType());
                        context.output(DAILY_ACTIVATE_UNITS, ucenterUser);
                    }
                }
            }
        }).uid("process");
        data.getSideOutput(DAILY_ACTIVATE_UNITS)
                .filter(o-> o.getLast_online_time()>= Util.getZeroPointTimestampsSecond())
                .uid("filter")
                .keyBy(UcenterUser::getId)
                .addSink(new DauKeyStateSink())
                .uid("DauKeyStateSink");
        env.execute("DailyActivateUnits");
    }
}
